#include "image_processor.h"

#include <condition_variable>
#include <filesystem>
#include <fstream>
#include <future>
#include <mutex>
#include <thread>

#include "cam_tele.h"
#include "dwarf_utils.h"
#include "elog.h"
#include "elog_file.h"
#include "ws_constant.h"
#include "ws_response.h"
#include "timer_utils.h"

sem_t sem_live_stacking;
std::mutex img_name_queue_mutex;
std::mutex stack_count_mutex;

std::condition_variable preprocessing_cv;
std::condition_variable align_stack_cv;
std::condition_variable sstretch_cv;
std::condition_variable backup_cv;

std::atomic<int> ImageProcessor::capture_light_display_source_ = STACKED;

// 星星叠图
int ImageProcessor::startLiveStackingPipeline(int img_to_take, const ExposureSetting& exp_setting, const IntValueSetting& gain_setting, int binning, const std::string& img_parent_path, nlohmann::json& shots_info, std::mutex& shots_info_mutex) {
    b_live_stacking = true;

    std::string file_name_timestamp = getTimeStamp();

    // 用于进度通知
    stack_count_ = 0;
    img_to_take_ = img_to_take;
    exp_setting_index_ = exp_setting.index;
    gain_setting_index_ = gain_setting.index;

    std::unique_lock<std::mutex> shots_info_lock(shots_info_mutex);
    target_name_ = shots_info.at("target");
    double target_ra = shots_info.at("RA");
    double target_dec = shots_info.at("DEC");
    shots_info_lock.unlock();

    std::ostringstream img_stacked_16_name;
    if (target_name_.empty())
        img_stacked_16_name << "/stacked-16_" << exp_setting.value << "s_" << gain_setting.value << "_" << file_name_timestamp << ".fits";
    else
        img_stacked_16_name << "/stacked-16_" << target_name_ << "_" << exp_setting.value << "s_" << gain_setting.value << "_" << file_name_timestamp << ".fits";

    std::string img_stacked_16_path = img_parent_path + img_stacked_16_name.str();
    std::string img_stacked_path = img_parent_path + img_stacked_name_;
    std::string shots_info_json_path = img_parent_path + shots_info_json_name_;

    std::queue<cv::Mat> img_queue;
    std::queue<cv::Mat> sstretch_queue;
    std::queue<cv::Mat> stack_queue;
    std::queue<int> stack_count_queue;

    std::mutex img_queue_mutex;
    std::mutex sstretch_queue_mutex;
    std::mutex stack_queue_mutex;

    int stack_count_pass = 0;  // 用于传递

    std::future<void> preprocess_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;
        time_checker timer;

        cv::Mat img_bad_pix_map;
        preprocessor_.readBadPixMap(dark_folder_, gain_setting.value, binning, img_bad_pix_map);

        while (1) {
            std::unique_lock<std::mutex> input_lock(img_name_queue_mutex);
            preprocessing_cv.wait(input_lock, [&] { return !img_name_queue_.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            std::string img_name = img_name_queue_.front();
            img_name_queue_.pop();
            img_name_queue_mutex.unlock();

            cv::Mat img_new;
            
            std::string img_new_path = img_parent_path + img_name;
            preprocessor_.calibrateLightFrame(img_new_path.c_str(), img_bad_pix_map, img_new);

            if (capture_light_display_source_ != STACKED) {
                if (!img_new.empty()) {
                    cv::Mat img_new_8b;
                    img_new.convertTo(img_new_8b, CV_8UC3, 1.0 / 256);
                    
                    matToByteArray(img_new_8b);
                }
            }

            timer.stop();
            timer.show_distance("Update stage ---- preprocess cost");

            std::unique_lock<std::mutex> output_lock(img_queue_mutex);
            align_stack_cv.wait(output_lock, [&] { return img_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            img_queue.push(img_new.clone());
            align_stack_cv.notify_one();
            output_lock.unlock();

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("preprocess thread EXIT");
    });

    std::future<void> align_stack_thread = std::async(std::launch::async, [&]() {
        time_checker timer;

        int img_count = 0;
        int stack_count = 0;

        cv::Mat img_primary;
        cv::Mat img_stack_all;
        cv::Mat img_stack_counter;

        std::thread save_unstretch_img_thread;

        while (1) {
            std::unique_lock<std::mutex> input_lock(img_queue_mutex);
            align_stack_cv.wait(input_lock, [&] { return !img_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_new = img_queue.front();
            img_queue.pop();
            align_stack_cv.notify_one();
            input_lock.unlock();

            cv::Mat img_stack;

            if (!img_new.empty()) {
                if (stack_count == 0) {
                    img_new.convertTo(img_primary, CV_8UC3, 1.0 / 256);
                    initLiveStackingImg(img_new, img_stack_all, img_stack_counter);

                    img_stack = img_new;

                    stack_count = 1;
                } else {
                    int ret = postprocessor_.alignAndStack(img_primary, img_new, img_stack, img_stack_all, img_stack_counter);
                    if (ret < 0) {
                        if (stack_count == 1) {
                            img_new.convertTo(img_primary, CV_8UC3, 1.0 / 256);
                            initLiveStackingImg(img_new, img_stack_all, img_stack_counter);
                        }
                        img_primary.release();
                    } else {
                        stack_count++;
                        img_primary.release();
                        postprocessor_.autoStretch(img_stack);
                    }
                }
            }

            timer.stop();
            timer.show_distance("Update stage ---- align and stack cost");

            std::unique_lock<std::mutex> output_lock(sstretch_queue_mutex);
            sstretch_cv.wait(output_lock, [&] { return sstretch_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            stack_count_pass = stack_count;
            sstretch_queue.push(img_stack.clone());
            sstretch_cv.notify_one();
            output_lock.unlock();

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("align and stack thread EXIT");
    });

    std::future<void> sstretch_thread = std::async(std::launch::async, [&]() {
        time_checker timer;

        int img_count = 0;

        std::thread save_unstretch_img_thread;

        while (1) {
            std::unique_lock<std::mutex> input_lock(sstretch_queue_mutex);
            sstretch_cv.wait(input_lock, [&] { return !sstretch_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_stack_before = sstretch_queue.front();
            sstretch_queue.pop();
            int stack_count = stack_count_pass;
            sstretch_cv.notify_one();
            input_lock.unlock();

            cv::Mat img_stack;

            if (!img_stack_before.empty()) {
                postprocessor_.sStretch(img_stack_before, img_stack);
                // img_stack = img_stack_before.clone();
            }

            // 当前在存图就更新队列，没存图就存图
            if (!img_stack.empty()) {
                cv::Mat img_stack_8b;
                img_stack.convertTo(img_stack_8b, CV_8UC3, 1.0 / 256);

                // 图传
                if (capture_light_display_source_ == STACKED)
                    matToByteArray(img_stack_8b);

                setAndNotifyStackCount(stack_count);

                std::unique_lock<std::mutex> output_lock(stack_queue_mutex);

                if (stack_queue.empty()) {
                    stack_queue.push(img_stack.clone());
                    stack_count_queue.push(stack_count);
                } else {
                    cv::Mat& img_stack_queue = stack_queue.front();
                    img_stack_queue = img_stack.clone();

                    int& img_stack_count = stack_count_queue.front();
                    img_stack_count = stack_count;
                }
                
                backup_cv.notify_one();
            } else {
                std::unique_lock<std::mutex> lock(shots_info_mutex);
                std::ofstream shots_info_json_file(shots_info_json_path);
                if (shots_info_json_file.is_open()) {
                    shots_info_json_file << std::setw(4) << shots_info << std::endl;
                    shots_info_json_file.close();
                } else {
                    log_e("shotsInfo.json OPEN FAILED!");
                }
            }

            timer.stop();
            timer.show_distance("Update stage ---- sstretch cost");

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("sstretch EXIT");
    });

    std::future<void> backup_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;
        time_checker timer;

        float time_all = 0;

        std::future<bool> save_stacked_16_img_thread;
        std::future<bool> save_stacked_img_thread;

        while (1) {
            std::unique_lock<std::mutex> input_lock(stack_queue_mutex);
            backup_cv.wait(input_lock, [&] { return !stack_queue.empty() || !b_live_stacking; });// 释放线程锁之前先判断lambda表达式，true的话不阻塞；false等待notify后再判断lambda表达式

            if (stack_queue.empty() && !b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_stack = stack_queue.front();
            int stack_count = stack_count_queue.front();
            stack_queue.pop();
            stack_count_queue.pop();
            backup_cv.notify_one();
            input_lock.unlock();
            
            if (!img_stack.empty()) {
                save_stacked_16_img_thread = std::async(std::launch::async, [&]() {
                    auto now = std::chrono::system_clock::now();
                    std::string fits_header_timestamp = getTimeStamp(now, "%Y-%m-%dT%H:%M:%S");

                    if (0 == file_manager_.convertToFits(exp_setting.value * stack_count, gain_setting.value, binning, target_ra, target_dec, target_name_.c_str(), fits_header_timestamp.c_str(), img_stack, img_stacked_16_path.c_str()))
                        return true;
                    else
                        return false;
                });

                save_stacked_img_thread = std::async(std::launch::async, [&]() {
                    cv::Mat img_stack_8b;
                    img_stack.convertTo(img_stack_8b, CV_8UC3, 1.0 / 256);
                    return cv::imwrite(img_stacked_path, img_stack_8b);
                });

                std::unique_lock<std::mutex> lock(shots_info_mutex);
                std::ofstream shots_info_json_file(shots_info_json_path);
                if (shots_info_json_file.is_open()) {
                    shots_info["shotsStacked"] = stack_count;
                    shots_info_json_file << std::setw(4) << shots_info << std::endl;

                    shots_info_json_file.close();
                } else {
                    log_e("shotsInfo.json OPEN FAILED!");
                }
            }

            if (save_stacked_16_img_thread.valid()) {
                int ret = save_stacked_16_img_thread.get();
            }

            if (save_stacked_img_thread.valid()) {
                int ret = save_stacked_img_thread.get();
            }

            img_count++;

            timer.stop();
            timer.show_distance("Update stage ---- backup cost");

            if (img_count >= img_to_take)
                break;
        }
        log_i("backup thread EXIT");
    });

    if (preprocess_thread.valid()) 
        preprocess_thread.wait();
    
    if (align_stack_thread.valid()) 
        align_stack_thread.wait();

    if (sstretch_thread.valid())
        sstretch_thread.wait();

    stopLiveStacking();
    
    if (backup_thread.valid()) 
        backup_thread.wait();

    std::thread sync_thread([&]() {
        system("sync");
    });
    sync_thread.detach();

    return 0;
}

// 太阳月亮叠图
int ImageProcessor::startLiveStackingSunMoonPipeline(int img_to_take, const ExposureSetting& exp_setting, const IntValueSetting& gain_setting, int binning, const std::string& img_parent_path, nlohmann::json& shots_info, std::mutex& shots_info_mutex) {
    b_live_stacking = true;

    std::string file_name_timestamp = getTimeStamp();

    // 用于进度通知
    stack_count_ = 0;
    img_to_take_ = img_to_take;
    exp_setting_index_ = exp_setting.index;
    gain_setting_index_ = gain_setting.index;
    
    std::unique_lock<std::mutex> shots_info_lock(shots_info_mutex);
    target_name_ = shots_info.at("target");
    double target_ra = shots_info.at("RA");
    double target_dec = shots_info.at("DEC");
    shots_info_lock.unlock();

    std::ostringstream img_stacked_16_name;
    if (target_name_.empty())
        img_stacked_16_name << "/stacked-16_" << exp_setting.value << "s_" << gain_setting.value << "_" << file_name_timestamp << ".fits";
    else
        img_stacked_16_name << "/stacked-16_" << target_name_ << "_" << exp_setting.value << "s_" << gain_setting.value << "_" << file_name_timestamp << ".fits";

    std::string img_stacked_16_path = img_parent_path + img_stacked_16_name.str();
    std::string img_stacked_path = img_parent_path + img_stacked_name_;
    std::string shots_info_json_path = img_parent_path + shots_info_json_name_;

    std::queue<cv::Mat> img_queue;
    std::queue<cv::Mat> stack_queue;
    std::queue<int> stack_count_queue;

    std::mutex img_queue_mutex;
    std::mutex stack_queue_mutex;

    std::future<void> preprocess_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;
        time_checker timer;

        while (1) {
            std::unique_lock<std::mutex> input_lock(img_name_queue_mutex);
            preprocessing_cv.wait(input_lock, [&] { return !img_name_queue_.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            std::string img_name = img_name_queue_.front();
            img_name_queue_.pop();
            img_name_queue_mutex.unlock();

            cv::Mat img_new;
            
            std::string img_new_path = img_parent_path + img_name;
            preprocessor_.readLightFrame(img_new_path.c_str(), img_new);

            if (capture_light_display_source_ != STACKED) {
                if (!img_new.empty()) {
                    cv::Mat img_new_8b;
                    img_new.convertTo(img_new_8b, CV_8UC3, 1.0 / 256);
                    
                    matToByteArray(img_new_8b);
                }
            }

            timer.stop();
            timer.show_distance("Update stage ---- preprocess cost");

            std::unique_lock<std::mutex> output_lock(img_queue_mutex);
            align_stack_cv.wait(output_lock, [&] { return img_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            img_queue.push(img_new.clone());
            align_stack_cv.notify_one();
            output_lock.unlock();

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("preprocess thread EXIT");
    });

    std::future<void> align_stack_thread = std::async(std::launch::async, [&]() {
        time_checker timer;

        int img_count = 0;
        int stack_count = 0;

        cv::Mat img_primary;
        cv::Mat img_stack_all;
        cv::Mat img_stack_counter;

        while (1) {
            std::unique_lock<std::mutex> input_lock(img_queue_mutex);
            align_stack_cv.wait(input_lock, [&] { return !img_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_new = img_queue.front();
            img_queue.pop();
            align_stack_cv.notify_one();
            input_lock.unlock();

            cv::Mat img_stack;

            if (!img_new.empty()) {
                if (stack_count == 0) {
                    img_primary = img_new.clone();
                    initLiveStackingImg(img_new, img_stack_all, img_stack_counter);

                    img_stack = img_new;

                    stack_count = 1;
                } else {
                    int ret = postprocessor_.solarSystemStack(img_primary, img_new, img_stack, img_stack_all, img_stack_counter);
                    if (ret < 0) {
                        if (stack_count == 1) {
                            img_primary = img_new.clone();
                            initLiveStackingImg(img_new, img_stack_all, img_stack_counter);
                        }
                    } else {
                        stack_count++;
                    }
                }
            }

            // 当前在存图就更新队列，没存图就存图
            if (!img_stack.empty()) {
                cv::Mat img_stack_8b;
                img_stack.convertTo(img_stack_8b, CV_8UC3, 1.0 / 256);

                // 图传
                if (capture_light_display_source_ == STACKED)
                    matToByteArray(img_stack_8b);

                setAndNotifyStackCount(stack_count);

                std::unique_lock<std::mutex> output_lock(stack_queue_mutex);

                if (stack_queue.empty()) {
                    stack_queue.push(img_stack.clone());
                    stack_count_queue.push(stack_count);
                } else {
                    cv::Mat& img_stack_queue = stack_queue.front();
                    img_stack_queue = img_stack.clone();
                    
                    int& img_stack_count = stack_count_queue.front();
                    img_stack_count = stack_count;
                }
                
                backup_cv.notify_one();
            } else {
                std::unique_lock<std::mutex> lock(shots_info_mutex);
                std::ofstream shots_info_json_file(shots_info_json_path);
                if (shots_info_json_file.is_open()) {
                    shots_info_json_file << std::setw(4) << shots_info << std::endl;
                    shots_info_json_file.close();
                } else {
                    log_e("shotsInfo.json OPEN FAILED!");
                }
            }

            timer.stop();
            timer.show_distance("Update stage ---- align and stack cost");

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("align and stack thread EXIT");
    });

    std::future<void> backup_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;
        time_checker timer;

        std::future<bool> save_stacked_16_img_thread;
        std::future<bool> save_stacked_img_thread;

        while (1) {
            std::unique_lock<std::mutex> input_lock(stack_queue_mutex);
            backup_cv.wait(input_lock, [&] { return !stack_queue.empty() || !b_live_stacking; });// 释放线程锁之前先判断lambda表达式，true的话不阻塞；false等待notify后再判断lambda表达式

            if (stack_queue.empty() && !b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_stack = stack_queue.front();
            int stack_count = stack_count_queue.front();
            stack_queue.pop();
            stack_count_queue.pop();
            backup_cv.notify_one();
            input_lock.unlock();
            
            if (!img_stack.empty()) {
                save_stacked_16_img_thread = std::async(std::launch::async, [&]() {
                    auto now = std::chrono::system_clock::now();
                    std::string fits_header_timestamp = getTimeStamp(now, "%Y-%m-%dT%H:%M:%S");

                    if (0 == file_manager_.convertToFits(exp_setting.value * stack_count, gain_setting.value, binning, target_ra, target_dec, target_name_.c_str(), fits_header_timestamp.c_str(), img_stack, img_stacked_16_path.c_str()))
                        return true;
                    else
                        return false;
                });

                save_stacked_img_thread = std::async(std::launch::async, [&]() {
                    cv::Mat img_stack_8b;
                    img_stack.convertTo(img_stack_8b, CV_8UC3, 1.0 / 256);
                    return cv::imwrite(img_stacked_path, img_stack_8b);
                });

                std::unique_lock<std::mutex> lock(shots_info_mutex);
                std::ofstream shots_info_json_file(shots_info_json_path);
                if (shots_info_json_file.is_open()) {
                    shots_info["shotsStacked"] = stack_count;
                    shots_info_json_file << std::setw(4) << shots_info << std::endl;

                    shots_info_json_file.close();
                } else {
                    log_e("shotsInfo.json OPEN FAILED!");
                }
            }

            if (save_stacked_16_img_thread.valid()) {
                int ret = save_stacked_16_img_thread.get();
            }

            if (save_stacked_img_thread.valid()) {
                int ret = save_stacked_img_thread.get();
            }

            img_count++;

            timer.stop();
            timer.show_distance("Update stage ---- backup cost");

            if (img_count >= img_to_take)
                break;
        }
        log_i("backup thread EXIT");
    });

    if (preprocess_thread.valid()) 
        preprocess_thread.wait();
    
    if (align_stack_thread.valid()) 
        align_stack_thread.wait();

    stopLiveStacking();
    
    if (backup_thread.valid()) 
        backup_thread.wait();

    std::thread sync_thread([&]() {
        system("sync");
    });
    sync_thread.detach();

    return 0;
}

// 使用线程锁保护获取的状态和主动通知的状态之间的顺序，确保后面的状态不会先发送
void ImageProcessor::setAndNotifyStackCount(int stack_count) {
    std::lock_guard<std::mutex> lock(stack_count_mutex);

    stack_count_ = stack_count;

    log_i("img_to_take_:%d, stack_count_:%d, exp_setting_index_:%d, gain_setting_index_:%d, target_name_:%s", img_to_take_, stack_count_.load(), exp_setting_index_, gain_setting_index_, target_name_.c_str());

    wsResNotifyProgressCaptureRawLiveStacking(CMD_NOTIFY_PROGRASS_CAPTURE_RAW_LIVE_STACKING, MODULE_NOTIFY, CMD_TYPE_NOTIFICATION, img_to_take_, STACKED_COUNT, 0, stack_count_, exp_setting_index_, gain_setting_index_, target_name_);
}

void ImageProcessor::notifyStackCount() {
    std::lock_guard<std::mutex> lock(stack_count_mutex);

    log_i("img_to_take_:%d, stack_count_:%d, exp_setting_index_:%d, gain_setting_index_:%d, target_name_:%s", img_to_take_, stack_count_.load(), exp_setting_index_, gain_setting_index_, target_name_.c_str());

    wsResNotifyProgressCaptureRawLiveStacking(CMD_NOTIFY_PROGRASS_CAPTURE_RAW_LIVE_STACKING, MODULE_NOTIFY, CMD_TYPE_NOTIFICATION, img_to_take_, STACKED_COUNT, 0, stack_count_, exp_setting_index_, gain_setting_index_, target_name_);
}

// 暗场叠加平均
int ImageProcessor::startLiveStackingPipeline(int img_to_take, const std::string& img_parent_path, double exp, int gain, int binning) {
    b_live_stacking = true;

    std::queue<cv::Mat> img_queue;
    std::queue<cv::Mat> stack_queue;

    std::mutex img_queue_mutex;
    std::mutex stack_queue_mutex;

    std::future<void> preprocess_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;
        time_checker timer;

        while(1) {
            std::unique_lock<std::mutex> input_lock(img_name_queue_mutex);
            preprocessing_cv.wait(input_lock, [&] { return !img_name_queue_.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            std::string img_name = img_name_queue_.front();
            img_name_queue_.pop();
            img_name_queue_mutex.unlock();

            std::string img_new_path = img_parent_path + img_name;
            
            cv::Mat img_new;
            ushort* img_new_data = NULL;
            file_manager_.convertTo16BitImg(img_new_path.c_str(), img_new_data, img_new);

            timer.stop();
            timer.show_distance("Update stage ---- preprocess cost");

            std::unique_lock<std::mutex> output_lock(img_queue_mutex);
            align_stack_cv.wait(output_lock, [&] { return img_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            img_queue.push(img_new.clone());
            align_stack_cv.notify_one();
            output_lock.unlock();

            xfree(img_new_data);

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("preprocess thread EXIT");
    });

    std::future<void> align_stack_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;
        int stack_count = 0;

        cv::Mat img_stack_all;

        time_checker timer;

        while(1) {
            std::unique_lock<std::mutex> input_lock(img_queue_mutex);
            align_stack_cv.wait(input_lock, [&] { return !img_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_new = img_queue.front();
            img_queue.pop();
            align_stack_cv.notify_one();
            input_lock.unlock();

            cv::Mat img_stack;

            if (!img_new.empty()) {
                if (stack_count == 0) {
                    stack_count = 1;

                    img_stack_all = img_new.clone();
                    img_stack_all.convertTo(img_stack_all, CV_32FC1);
                    img_stack = img_new;
                } else {
                    stack_count++;

                    img_stack_all += img_new;
                    img_stack = img_stack_all / (float)stack_count;
                    img_stack.convertTo(img_stack, CV_16UC1);
                }
            }

            timer.stop();
            timer.show_distance("Update stage ---- align and stack cost");

            std::unique_lock<std::mutex> output_lock(stack_queue_mutex);
            backup_cv.wait(output_lock, [&] { return stack_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            stack_queue.push(img_stack.clone());
            backup_cv.notify_one();
            output_lock.unlock();

            img_count++;

            if (img_count >= img_to_take)
                break;
        }
        log_i("align and stack thread EXIT");
    });

    std::future<void> backup_thread = std::async(std::launch::async, [&]() {
        int img_count = 0;

        time_checker timer;

        while(1) {
            std::unique_lock<std::mutex> input_lock(stack_queue_mutex);
            backup_cv.wait(input_lock, [&] { return !stack_queue.empty() || !b_live_stacking; });

            if (!b_live_stacking) 
                break;

            timer.start();

            cv::Mat img_stack = stack_queue.front();
            stack_queue.pop();
            backup_cv.notify_one();
            input_lock.unlock();

            // 图传
            if (!img_stack.empty()) {
                cv::Mat img_stack_8b;
                img_stack.convertTo(img_stack_8b, CV_8UC1, 1.0 / 256);

                matToByteArray(img_stack_8b);
            }

            timer.stop();
            timer.show_distance("Update stage ---- send dark cost");

            img_count++;

            if (img_count >= img_to_take) {
                if (img_stack.empty())// 只要img_stack不为空，就能保存，不管叠了几张
                    break;

                if (!std::filesystem::exists(dark_folder_))
                    std::filesystem::create_directory(dark_folder_);

                cv::Mat img_bad_pix;
                preprocessor_.generateBadPixMap(img_stack, img_bad_pix);

                std::ostringstream img_dark_path;
                img_dark_path << dark_folder_ << "exp_" << std::fixed << std::setprecision(6) << exp << "_gain_" << gain << "_bin_" << binning + 1 << ".png";
                cv::imwrite(img_dark_path.str().c_str(), img_stack);

                if(!img_bad_pix.empty()) {
                    std::ostringstream img_bad_pix_map_path;
                    img_bad_pix_map_path << dark_folder_ << "badPix_exp_" << std::fixed << std::setprecision(6) << exp << "_gain_" << gain << "_bin_" << binning + 1 << "_thre_0.5.png";
                    cv::imwrite(img_bad_pix_map_path.str().c_str(), img_bad_pix);
                }

                std::thread sync_thread([&]() {
                    system("sync");
                });
                sync_thread.detach();

                timer.stop();
                timer.show_distance("Update stage ---- backup cost");

                break;
            }
        }
        log_i("backup thread EXIT");
    });

    if (preprocess_thread.valid()) 
        preprocess_thread.wait();

    if (align_stack_thread.valid()) 
        align_stack_thread.wait();
    
    if (backup_thread.valid()) 
        backup_thread.wait();

    return 0;
}

void ImageProcessor::stopLiveStacking() {
    b_live_stacking = false;

    preprocessing_cv.notify_all();
    align_stack_cv.notify_all();
    sstretch_cv.notify_all();
    backup_cv.notify_all();
}

void ImageProcessor::addImg(const std::string& img_name) { 
    img_name_queue_mutex.lock();
    img_name_queue_.push(img_name); 
    img_name_queue_mutex.unlock();

    preprocessing_cv.notify_all();
}

void ImageProcessor::popAllImg() { 
    std::queue<std::string> emptyQueue;
    img_name_queue_.swap(emptyQueue);
}

void ImageProcessor::initLiveStackingImg(const cv::Mat& img_new, cv::Mat& img_stack_all, cv::Mat& img_stack_counter) {
    img_stack_all.release();
    img_stack_counter.release();

    img_stack_all = img_new.clone();
    img_stack_all.convertTo(img_stack_all, CV_32FC3);
    img_stack_counter = cv::Mat(img_new.rows, img_new.cols, CV_16UC3, cv::Scalar(1, 1, 1));
}

int ImageProcessor::sem_timedwait_millsecs(sem_t *sem, long msecs) {
	struct timespec ts;
	clock_gettime(CLOCK_REALTIME, &ts);
	long secs = msecs / 1000;
	msecs = msecs % 1000;

	long add = 0;
	msecs = msecs * 1000 * 1000 + ts.tv_nsec;
	add = msecs / (1000 * 1000 * 1000);
	ts.tv_sec += (add + secs);
	ts.tv_nsec = msecs % (1000 * 1000 * 1000);

	return sem_timedwait(sem, &ts);
}
